package com.mjf.hotitems_analysis

import java.sql.Timestamp

import com.mjf.dim.UserBehavior
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.{EnvironmentSettings, Slide, Table}

object HotItemsWithSql {
  def main(args: Array[String]): Unit = {

    // 创建流处理执行环境
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    // 定义时间语义
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)

    // 从文件读取数据
    val inputStream: DataStream[String] = env.readTextFile("D:\\coding\\idea\\UserBehaviorAnalysis\\HotItemsAnalysis\\src\\main\\resources\\UserBehavior.csv")

    // 将数据转换为样例类，并且提取timestamp定义watermark
    val dataStream: DataStream[UserBehavior] = inputStream.map {
      line =>
        val dataArray: Array[String] = line.split(",")
        UserBehavior(dataArray(0).toLong, dataArray(1).toLong, dataArray(2).toInt, dataArray(3), dataArray(4).toLong)
    }.assignAscendingTimestamps(_.timestamp * 1000L)  // 数据是有序的，不需要定义延迟处理乱序数据

    // 要调用TableApi,先创建表执行环境
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance()
      .useBlinkPlanner()
      .inStreamingMode()
      .build()

    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(env, settings)

    // 将DataStream注册成表，提取需要的字段进行处理
    tableEnv.createTemporaryView("data_table", dataStream, 'itemId, 'behavior, 'timestamp.rowtime as 'ts)

    val resultTable: Table = tableEnv.sqlQuery(
      """
        |select
        |   *
        |from
        |   (
        |   select
        |       *,
        |       row_number() over(partition by windowEnd order by cnt desc) as rn
        |   from
        |       (
        |       select
        |           itemId,
        |           count(itemId) as cnt,
        |           hop_end(ts, interval '5' minute, interval '1' hour) as windowEnd
        |       from
        |           data_table
        |       where
        |           behavior = 'pv'
        |       group by
        |           itemId,
        |           hop(ts, interval '5' minute, interval '1' hour)
        |       ) as a
        |   ) as a
        |where
        |   rn <= 5
        |""".stripMargin)

    resultTable.toRetractStream[(Long, Long, Timestamp, Long)].print("result")

    env.execute("HotItemsWithSql")

  }
}
